
#include <google/protobuf/descriptor.h>
#include <google/protobuf/message.h>
#include <google/protobuf/service.h>

#include "../../common/error_code.h"
#include "../../common/log.h"
#include "../../common/req_id_util.h"
#include "../coder/tinypb_protocol.cc"
#include "../tcp/tcp_client.h"
#include "rpc_channel.h"
#include "rpc_controller.h"

namespace rocket {

RpcChannel::RpcChannel(NetAddr::s_ptr peer_addr) : m_peer_addr(peer_addr) {
    m_client = std::make_shared<TcpClient>(m_peer_addr);
}

RpcChannel::~RpcChannel() { INFOLOG("~RcpChannel()"); }

void RpcChannel::CallMethod(const google::protobuf::MethodDescriptor *method,
                            google::protobuf::RpcController *controller,
                            const google::protobuf::Message *request,
                            google::protobuf::Message *response,
                            google::protobuf::Closure *done) {
    std::shared_ptr<rocket::TinyPBProtocol> req_protocol =
        std::make_shared<rocket::TinyPBProtocol>();

    RpcController *my_controller = dynamic_cast<RpcController *>(controller);
    if (my_controller == NULL) {
        ERRORLOG("failed callmethod, RpcController convert error");
        return;
    }

    if (my_controller->GetReqId().empty()) {
        req_protocol->m_req_id = ReqIdUtil::GenReqId();
        my_controller->SetReqId(req_protocol->m_req_id);
    } else {
        req_protocol->m_req_id = my_controller->GetReqId();
    }

    req_protocol->m_method_name = method->full_name();
    INFOLOG("%s | call method name [%s]", req_protocol->m_req_id.c_str(),
            req_protocol->m_method_name.c_str());

    if (!m_is_init) {
        std::string err_info = "RpcChannel not init";
        my_controller->SetError(ERROR_RPC_CHANNEL_INIT, err_info);
        ERRORLOG("%s | %s, RpcChannel not init", req_protocol->m_req_id.c_str(),
                 err_info.c_str());
        return;
    }

    // request serialization
    if (request->SerializeToString(&(req_protocol->m_pb_data)) == false) {
        std::string err_info = "fail to serialize";
        my_controller->SetError(ERROR_FAILED_SERIALIZE, err_info);
        ERRORLOG("%s | %s, oerigin request [%s]",
                 req_protocol->m_req_id.c_str(), err_info.c_str(),
                 request->ShortDebugString());
    }

    s_ptr channel = shared_from_this();

    auto timeOutDone = [my_controller, &channel](){
        my_controller->StartCancel();
        my_controller->SetError(ERROR_RPC_CALL_TIMEOUT, "rpc call timeout " + std::to_string(my_controller->GetTimeout()));
        if (channel->getClosure()) {
            channel->getClosure()->Run();
        }
        channel.reset();
    };
    m_timer_event = std::make_shared<TimerEvent>(my_controller->GetTimeout(), false, timeOutDone);
    m_client->addTimerEvent(m_timer_event);

    auto writeDone = [req_protocol, channel](AbstractProtocol::s_ptr msg) {
        INFOLOG("%s | rpc send request success, call method_name[%s], peer addr[%s], local addr[%s]",
                req_protocol->m_req_id.c_str(),
                req_protocol->m_method_name.c_str(),
                channel->getTcpClient()->getPeerAddr()->toString().c_str(),
                channel->getTcpClient()->getLocalAddr()->toString().c_str());
    };
    auto readDone = [channel, my_controller](AbstractProtocol::s_ptr msg) mutable {
        // RPC 完成，取消定时任务
        channel->getTimerEvent()->setCancelled(true);
        
        std::shared_ptr<rocket::TinyPBProtocol> rsp_protocol =
            std::dynamic_pointer_cast<rocket::TinyPBProtocol>(msg);
        INFOLOG("%s | success get rpc response, call method name[%s], peer addr[%s], local addr[%s]",
                rsp_protocol->m_req_id.c_str(),
                rsp_protocol->m_method_name.c_str(),
                channel->getTcpClient()->getPeerAddr()->toString().c_str(),
                channel->getTcpClient()->getLocalAddr()->toString().c_str());

        if (!(channel->getResponse()->ParseFromString(
                rsp_protocol->m_pb_data))) {
            ERRORLOG("%s | deserialize error", rsp_protocol->m_req_id.c_str());
            my_controller->SetError(ERROR_FAILED_DESERIALIZE,
                                    "deserialize error");
            return;
        }

        if (rsp_protocol->m_err_code != 0) {
            ERRORLOG("%s | call rpc method [%s] failied, error code[%d], error "
                     "info[%s]",
                     rsp_protocol->m_req_id.c_str(),
                     rsp_protocol->m_method_name.c_str(),
                     rsp_protocol->m_err_code,
                     rsp_protocol->m_err_info.c_str());
            my_controller->SetError(rsp_protocol->m_err_code,
                                    rsp_protocol->m_err_info);
            return;
        }

        INFOLOG("%s | call rpc success, call method name[%s], peer addr[%s], local addr[%s]",
                    rsp_protocol->m_req_id.c_str(), 
                    rsp_protocol->m_method_name.c_str(),
                    channel->getTcpClient()->getPeerAddr()->toString().c_str(),
                    channel->getTcpClient()->getLocalAddr()->toString().c_str());

        if (!my_controller->IsCanceled() && channel->getClosure()) {
            channel->getClosure()->Run();
        }
        channel.reset();
    };
    auto connectDone = [req_protocol, writeDone, readDone, channel, my_controller]() {
        if (channel->getTcpClient()->getConnectErrorCode() != 0) {
            my_controller->SetError(channel->getTcpClient()->getConnectErrorCode(), channel->getTcpClient()->getConnectErrorInfo());
            ERRORLOG("%s | connect error, error code[%d], error info[%s], peer addr[%s]", 
                req_protocol->m_req_id.c_str(), 
                my_controller->GetErrorCode(),
                my_controller->GetErrorInfo().c_str(), channel->getTcpClient()->getPeerAddr()->toString().c_str());
            return ;
        }

        channel->getTcpClient()->writeMessage(req_protocol, writeDone);
        channel->getTcpClient()->readMessage(req_protocol->m_req_id, readDone);
    };
    m_client->connect(connectDone);
}

void RpcChannel::Init(controller_s_ptr controller, message_s_ptr req,
                      message_s_ptr res, closure_s_ptr done) {
    if (m_is_init) {
        return;
    }
    m_controller = controller;
    m_request = req;
    m_response = res;
    m_closure = done;
    m_is_init = true;
}
google::protobuf::RpcController *RpcChannel::getController() {
    return m_controller.get();
}

google::protobuf::Message *RpcChannel::getRequest() { 
    return m_request.get(); 
}

google::protobuf::Message *RpcChannel::getResponse() {
    return m_response.get();
}

google::protobuf::Closure *RpcChannel::getClosure() { 
    return m_closure.get(); 
}

TcpClient *RpcChannel::getTcpClient() { 
    return m_client.get(); 
}

TimerEvent::s_ptr RpcChannel::getTimerEvent() {
    return m_timer_event;
}


} // namespace rocket
